package skewreduce.pig;

import java.io.DataInput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URL;
import java.nio.ByteOrder;
import java.util.Arrays;
import java.util.Map;

import eagg.lib.EndianConvertor;

import org.apache.hadoop.io.DataInputBuffer;

import org.apache.pig.ExecType;
import org.apache.pig.LoadFunc;
import org.apache.pig.PigException;
import org.apache.pig.StoreFunc;
import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.io.BufferedPositionedInputStream;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;

public abstract class EaggLoadBase implements LoadFunc, StoreFunc {
    protected static enum TYPE {
        UNKNOWN (0,DataType.UNKNOWN),
        BYTE (1,DataType.BYTE),
        UNSIGNEDBYTE(1,DataType.INTEGER),
        SHORT (2,DataType.INTEGER),
        UNSIGNEDSHORT (2,DataType.INTEGER),
        INT (4,DataType.INTEGER),
        LONG (8,DataType.LONG),
        FLOAT (4,DataType.FLOAT),
        DOUBLE (8,DataType.DOUBLE),
        BYTEARRAY (4,DataType.BYTEARRAY);

        private final int  sz;
        private final byte pigType;

        TYPE(int sz,byte t) { this.sz = sz; pigType = t; }

        public int getSize() { return sz; }
        public byte getPigDataType() { return pigType; }
    }

    protected Object getObject(TYPE t,DataInput in) throws IOException {
        switch ( t ) {
            case BYTE:
                return new Byte(in.readByte());
            case UNSIGNEDBYTE:
                return new Integer(in.readUnsignedByte());
            case SHORT:
                return new Integer(in.readShort());
            case UNSIGNEDSHORT:
                return new Integer(in.readUnsignedShort());
            case INT:
                return new Integer(in.readInt());
            case LONG:
                return new Long(in.readLong());
            case FLOAT:
                return new Float(in.readFloat());
            case DOUBLE:
                return new Double(in.readDouble());
            case BYTEARRAY:
                {
                int sz = in.readInt();
                byte[] data = new byte[sz];
                in.readFully(data);
                return data;
                }
            default:
                throw new IOException("Unknown data format: "+t);
        }
    }

    /**
     * schema for the input
     */
    protected Schema     schema;
    protected TYPE[]     types;
    protected int        minRecSize;
    private TupleFactory tupleFactory = TupleFactory.getInstance();
    private   boolean    littleEndian;

    private DataInputBuffer buffer = new DataInputBuffer();
    private EndianConvertor convertor = new EndianConvertor();

    public EaggLoadBase() {}

    public EaggLoadBase(String fmt) {
        parseSchemaString(fmt);
    }

    public Schema getSchema() { return schema; }

    protected int getMinRecordSize() { return minRecSize; }

    /**
     * parse given format string and construct schema structure
     */
    protected void parseSchemaString(String fmt) {
        ByteOrder byteOrder = ByteOrder.BIG_ENDIAN;
        switch ( fmt.charAt(0) ) {
            case '@': case '=':
                byteOrder = ByteOrder.nativeOrder();
                break;
            case '<':
                byteOrder = ByteOrder.LITTLE_ENDIAN;
                break;
        }
        littleEndian = ( byteOrder == ByteOrder.LITTLE_ENDIAN );

        int i = "@=<>!".indexOf( fmt.charAt(0) );
        String[] fields = ( ( i < 0 ) ? fmt : fmt.substring(1) ).toUpperCase().split(",");

        schema = new Schema();
        types = new TYPE[fields.length];
        int fldIdx = 0;

        for ( String field : fields ) {
            String[] xxx = field.trim().split(":");
            String alias = null;
            String valueType = null;
            if ( xxx.length == 2 ) {
                alias = xxx[0].toLowerCase(); valueType = xxx[1];
            } else if ( xxx.length == 1 ) {
                valueType = xxx[0];
            } else {
                throw new IllegalArgumentException("wrong schema: "+field);
            }

            TYPE type = TYPE.valueOf(TYPE.class, valueType);
            schema.add(new Schema.FieldSchema(alias,type.getPigDataType()));
            types[fldIdx++] = type;
            minRecSize += type.getSize();
        }
    }

    private DataInput wrap(DataInput in) {
        return ( littleEndian ) ? convertor.set(in) : in;
    }

    protected Tuple createTuple(DataInput in) throws IOException {
        DataInput input = wrap(in);
        Tuple t = tupleFactory.newTuple( types.length );
        for ( int i = 0; i < types.length; ++i ) {
            t.set(i, getObject(types[i], input) );
        }
        return t;
    }

    /**
     * Cast data from bytes to integer value.  
     * @param b byte array to be cast.
     * @return Integer value.
     * @throws IOException if the value cannot be cast.
     */
    public Integer bytesToInteger(byte[] b) throws IOException {
        buffer.reset(b,b.length);
        DataInput in = wrap(buffer);
        int r;
        switch ( b.length ) {
            case 4: r = in.readInt(); break;
            case 8: r = (int)in.readLong(); break;
            case 2: r = (int)in.readShort(); break;
            case 1: r = (int)in.readByte(); break;
            default:
                throw new IOException("Invalid size: "+b.length);
        }
        return new Integer(r);
    }

    /**
     * Cast data from bytes to long value.  
     * @param b byte array to be cast.
     * @return Long value.
     * @throws IOException if the value cannot be cast.
     */
    public Long bytesToLong(byte[] b) throws IOException {
        buffer.reset(b,b.length);
        DataInput in = wrap(buffer);
        long r;
        switch ( b.length ) {
            case 8: r = in.readLong(); break;
            case 4: r = (long)in.readInt(); break;
            case 2: r = (long)in.readShort(); break;
            case 1: r = (long)in.readByte(); break;
            default:
                throw new IOException("Invalid size: "+b.length);
        }
        return new Long(r);
    }

    /**
     * Cast data from bytes to float value.  
     * @param b byte array to be cast.
     * @return Float value.
     * @throws IOException if the value cannot be cast.
     */
    public Float bytesToFloat(byte[] b) throws IOException {
        buffer.reset(b,b.length);
        DataInput in = wrap(buffer);
        float r;
        switch ( b.length ) {
            case 4: r = in.readFloat(); break;
            case 8: r = (float)in.readDouble(); break;
            default:
                throw new IOException("Invalid size: "+b.length);
        }
        return new Float(r);
    }

    /**
     * Cast data from bytes to double value.  
     * @param b byte array to be cast.
     * @return Double value.
     * @throws IOException if the value cannot be cast.
     */
    public Double bytesToDouble(byte[] b) throws IOException {
        buffer.reset(b,b.length);
        DataInput in = wrap(buffer);
        double r;
        switch ( b.length ) {
            case 8: r = in.readDouble(); break;
            case 4: r = (double)in.readFloat(); break;
            default:
                throw new IOException("Invalid size: "+b.length);
        }
        return new Double(r);
    }

    /**
     * Cast data from bytes to chararray value.  
     * @param b byte array to be cast.
     * @return String value.
     * @throws IOException if the value cannot be cast.
     */
    public String bytesToCharArray(byte[] b) throws IOException {
        throw new IOException("Unsupported");
    }

    /**
     * Cast data from bytes to map value.  
     * @param b byte array to be cast.
     * @return Map value.
     * @throws IOException if the value cannot be cast.
     */
    public Map<String, Object> bytesToMap(byte[] b) throws IOException {
        throw new IOException("Unsupported");
    }

    /**
     * Cast data from bytes to tuple value.  
     * @param b byte array to be cast.
     * @return Tuple value.
     * @throws IOException if the value cannot be cast.
     */
    public Tuple bytesToTuple(byte[] b) throws IOException {
        throw new IOException("Unsupported");
    }

    /**
     * Cast data from bytes to bag value.  
     * @param b byte array to be cast.
     * @return Bag value.
     * @throws IOException if the value cannot be cast.
     */
    public DataBag bytesToBag(byte[] b) throws IOException {
        throw new IOException("Unsupported");
    }

    public LoadFunc.RequiredFieldResponse fieldsToRead(LoadFunc.RequiredFieldList requiredFieldList) throws FrontendException {
        return new LoadFunc.RequiredFieldResponse(false);
    }

    public void fieldsToRead(Schema schema) {}

    public Schema determineSchema(String fileName, ExecType execType, DataStorage storage) throws IOException {
        return schema;
    }

    public boolean equals(Object o) {
        if ( this == o ) return true;
        if ( o instanceof EaggLoadBase ) {
            return Arrays.equals(types, ((EaggLoadBase)o).types);
        }
        return false;
    }

    ///////////////////////////////////////////////////////////////////////////
    // write implementation
    private DataOutputStream out;

    public void bindTo(OutputStream os) throws IOException {
        out = new DataOutputStream(os);
    }

    private void putField(TYPE t,Object o) throws IOException {
        // get next
        if ( o instanceof Number ) {
            Number n = (Number)o;
            switch ( t ) {
                case UNSIGNEDBYTE:
                case BYTE: out.writeByte(n.intValue()); break;
                case UNSIGNEDSHORT:
                case SHORT: out.writeShort(n.intValue()); break;
                case INT: out.writeInt(n.intValue()); break;
                case LONG: out.writeLong(n.longValue()); break;
                case FLOAT: out.writeFloat(n.floatValue()); break;
                case DOUBLE: out.writeDouble(n.doubleValue()); break;
                default: {
                    int errCode = 2108;
                    String msg = "Could not determine data type of field: " + o;
                    throw new ExecException(msg, errCode, PigException.BUG);
                }
            }
        } else {
            switch ( t ) {
                case BYTEARRAY: {
                    out.writeInt(((byte[])o).length);
                    out.write((byte[])o);
                    break;
                    }
                default: {
                    int errCode = 2108;
                    String msg = "Could not determine data type of field: " + o;
                    throw new ExecException(msg, errCode, PigException.BUG);
                }
            }
        }
    }

    public void putNext(Tuple f) throws IOException {
        for ( int i = 0; i < types.length; ++i ) {
            Object field;
            try {
                field = f.get(i);
            } catch ( ExecException ee) {
                throw ee;
            }
            putField(types[i],field);
        }
    }
    public void finish() throws IOException {
        out.close();
    }

    public Class getStorePreparationClass() throws IOException {
        return null;
    }
}
